package alfred.mdp.client;

import alfred.mdp.messages.client.Reply;
import alfred.mdp.messages.client.Request;
import org.zeromq.ZContext;
import org.zeromq.ZDispatcher;
import org.zeromq.ZMQ;
import org.zeromq.ZMsg;

import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import static alfred.mdp.utils.Utils.asByteArray;

/**
 */
public class Client {
    private final ZMQ.Socket socket;
    private final String id;
    private final ZDispatcher.ZSender sender;
    private final ZDispatcher dispatcher;
    private ZMsg response;
    private CountDownLatch receiverLatch;

    public Client(ZContext context, ZDispatcher dispatcher, String endpoint) {
        this.id = UUID.randomUUID().toString();
        this.socket = context.createSocket(ZMQ.DEALER);
        this.socket.setIdentity(asByteArray(id));
        this.socket.connect(endpoint);
        this.sender = new ZDispatcher.ZSender();
        this.dispatcher = dispatcher;
        this.dispatcher.registerHandler(socket, new ZDispatcher.ZMessageHandler() {
            @Override
            public void handleMessage(ZDispatcher.ZSender sender, ZMsg msg) {
                response = new Reply(msg).getBody();
                receiverLatch.countDown();
            }
        }, sender);
    }

    public ZMsg request(String service, ZMsg message, TimeUnit unit, long timeout) {
        try {
            receiverLatch = new CountDownLatch(1);
            sender.send(new Request(service, message).getZMsg());
            if (receiverLatch.await(timeout, unit)) {
                return response;
            }
        } catch (InterruptedException e) {
        }
        throw new ClientTimeoutException();
    }

    public void disconnect() {
        dispatcher.unregisterHandler(socket);
        this.socket.setLinger(0);
        this.socket.close();
    }

    public String getId() {
        return id;
    }

}